package com.stone.tech.cdc.config;

import com.google.common.collect.Lists;
import com.stone.tech.cdc.model.DatabaseInfo;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;


@Configuration
@ConfigurationProperties(prefix = "debezium")
public class DebeziumMysqlConfiguration {

    @NestedConfigurationProperty
    private List<DatabaseInfo> databases;

    public void setDatabases(List<DatabaseInfo> databases) {
        this.databases = databases;
    }

    // @Bean
    public List<String> allTables() {
        List<String> retList = Lists.newLinkedList();
        for (DatabaseInfo database : databases) {
            retList.addAll(database.getTables());
        }
        return retList;
    }

    @Bean
    public List<Properties> debeziumMysqlProperties() {
        List<Properties> propList = Lists.newLinkedList();
        String osname = System.getProperty("os.name");
        String path = "/data/debezium/mysql";
        if (osname.startsWith("Windows")) {
            path = "D:/tmp/debezium/mysql";
        }
        for (DatabaseInfo dbi : databases) {

            dbi.setOffsetPath(path + dbi.getOffsetPath());
            dbi.setHistoryPath(path + dbi.getHistoryPath());

            Properties props = new Properties();
            props.setProperty("name", dbi.getName());
            props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
            props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
            props.setProperty("offset.storage.file.filename", dbi.getOffsetPath());
            props.setProperty("offset.flush.interval.ms", "600000");
            props.setProperty("database.hostname", dbi.getHost());
            props.setProperty("database.port", dbi.getPort());
            props.setProperty("database.user", dbi.getUsername());
            props.setProperty("database.password", dbi.getPassword());
            props.setProperty("database.server.id", dbi.getServerId());
            props.setProperty("database.server.name", "my_mysql_connector" + dbi.getName());
            props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
            props.setProperty("database.history.file.filename", dbi.getHistoryPath());
            String tableList = dbi.getTables().stream()
                    .map(item -> item.indexOf("&") > -1 ? item.substring(0, item.indexOf("&")) : item)
                    .collect(Collectors.joining(","));
            props.setProperty("table.include.list", tableList);
            propList.add(props);
        }
        return propList;
    }
}
